package com.shujia.road


import java.sql.{Connection, PreparedStatement}

import com.alibaba.fastjson.{JSON, JSONObject}
import com.shujia.deploy.HaveBeenSupervisedCheck.env
import com.shujia.flink.FlinkTool
import com.shujia.util.{CarUtil, JdbcUtil}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

/**
  * 1.3 实时统计道路拥堵情况-
  * 实时计算每个道路平均速度，计算最近5分钟每个一分钟计算一次
  * 输出结果
  * 道路编号, 计算时间（窗口结束时间），平均车速，总的车辆
  *
  */
object RealTimeRoadAvgSpeed extends FlinkTool {
  override def run(args: Array[String]): Unit = {

    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val carsDS: DataStream[String] = CarUtil.loadKafkaData(env)


    //解析数据

    val kvDS: DataStream[(Long, Double, Long)] = carsDS.map(car => {
      val carJson: JSONObject = JSON.parseObject(car)
      val road_id: Long = carJson.getLong("road_id")
      val speed: Double = carJson.getDouble("speed")
      val time: Long = carJson.getLong("time")

      (road_id, speed, time * 1000)
    })

    //设置时间字段和水位线

    val assDS: DataStream[(Long, Double, Long)] = kvDS.assignTimestampsAndWatermarks(
      //指定时间戳字段和数据最大乱序时间
      new BoundedOutOfOrdernessTimestampExtractor[(Long, Double, Long)](Time.seconds(10)) {
        override def extractTimestamp(element: (Long, Double, Long)): Long = element._3
      }
    )

    //划分窗口
    val windowDS: WindowedStream[(Long, Double, Long), Long, TimeWindow] = assDS
      .keyBy(_._1)
      .timeWindow(Time.minutes(5), Time.minutes(1))


    /**
      * 在窗口内计算平均值车速和总的车辆
      */
    val avgSpeedDS: DataStream[(Long, Long, Double, Long)] = windowDS.process(new ProcessWindowFunction[(Long, Double, Long), (Long, Long, Double, Long), Long, TimeWindow] {
      override def process(key: Long,
                           context: Context,
                           elements: Iterable[(Long, Double, Long)], //窗口内所有的数据
                           out: Collector[(Long, Long, Double, Long)]): Unit = {

        //1、计算平均车速

        //总的车辆
        var sumCar = 0L
        //总的车速
        var sumSpeed = 0.0

        for (car <- elements) {
          sumCar += 1
          sumSpeed += car._2
        }

        //平均车速
        val avgSpeed: Double = sumSpeed / sumCar


        //获取窗口的结束时间
        val windowEndTime: Long = context.window.getEnd

        //将数据发送到下游
        out.collect((key, windowEndTime, avgSpeed, sumCar))
      }
    })

    /**
      * 将计算结果保存到mysql中
      *
      */

    avgSpeedDS.addSink(new RichSinkFunction[(Long, Long, Double, Long)] {
      var con: Connection = _

      override def open(parameters: Configuration): Unit = {
        con = JdbcUtil.getConnection
      }

      override def close(): Unit = {
        con.close()
      }

      override def invoke(value: (Long, Long, Double, Long), context: SinkFunction.Context[_]): Unit = {

        val stat: PreparedStatement = con.prepareStatement("replace into real_time_road_avg_speed(road_id,window_end_time,avg_speed,sum_car) values(?,?,?,?)")

        stat.setLong(1, value._1)
        stat.setLong(2, value._2)
        stat.setDouble(3, value._3)
        stat.setLong(4, value._4)

        stat.execute()


      }
    })


    env.execute("RealTimeRoadAvgSpeed$")


  }
}
